﻿
'a distributor is working on breaking up tasks from one queue to other queues
'[input] -> distributor -> [planet1 Q] [planet2 Q] ...

Imports std

Partial Public MustInherit Class distributor
    Inherits consumer

    Private ReadOnly _targetQueues As vector(Of String) = Nothing
    Private OVERALL_TIME_MS As Int64 = 0
    Private WAIT_TARGET_JOBS_TIME_MS As Int64 = 0

    Public Sub New()
        _targetQueues = New vector(Of String)()
    End Sub

    Private Shadows Sub initial(ByVal targetQueues As String)
        assert(Not isemptystring(targetQueues))
        strsepar(targetQueues, targetQueuesSeparator, _targetQueues, True)
        For i As Int64 = 0 To _targetQueues.size() - 1
            kickLeftRightBlank(_targetQueues(i))
        Next
        assert(Not _targetQueues Is Nothing AndAlso _targetQueues.size() > 0)
    End Sub

    Protected Shared Function make_pair(ByVal index As UInt32, ByVal value As ijob) As pair(Of UInt32, ijob)
        Return New pair(Of UInt32, ijob) With {.first = index, .second = value}
    End Function

    Protected Function targetQueues() As vector(Of String)
        Return _targetQueues
    End Function

    Protected Function targetQueue(ByVal i As UInt32) As jobqueue
        Return jobqueues.at(targetQueues()(i))
    End Function

    Protected Function hashcode(Of T)(ByVal i As T) As UInt32
        Return _ijob.hashcode(i, targetQueues.size())
    End Function

    Protected Function nextQueueIndex(ByVal i As UInt32) As UInt32
        If _inc(i) >= targetQueues().size() Then
            Return 0
        Else
            Return i
        End If
    End Function

    Protected MustOverride Function distribute(ByVal inputJob As ijob,
                                               ByVal targetJobs As vector(Of pair(Of UInt32, ijob)),
                                               ByRef waitall As Boolean) As Boolean

    Protected Overridable Function aggregate(ByVal targetJobs() As indexedJob, ByVal inputJob As ijob) As Boolean
        Return assert(False)
    End Function

    Protected Overridable Function aggregate(ByVal targetJobs() As indexedJob,
                                             ByVal jobqueues() As jobqueue,
                                             ByVal queue_indexes() As UInt32,
                                             ByVal inputJob As ijob) As Boolean
        Return aggregate(targetJobs, inputJob)
    End Function

    Private Class targets_t
        Public ReadOnly ecs() As eventComb = Nothing
        Public ReadOnly indexedJobs() As indexedJob = Nothing
        Public ReadOnly jobqueues() As jobqueue = Nothing
        Public ReadOnly queue_indexes() As UInt32 = Nothing

        Public Sub New(ByVal size As Int64)
            size -= 1
            ReDim ecs(size)
            ReDim indexedJobs(size)
            ReDim jobqueues(size)
            ReDim queue_indexes(size)
        End Sub

        Public Function size() As Int64
            assert(ecs.Length() = indexedJobs.Length() AndAlso
                   indexedJobs.Length() = jobqueues.Length())
            Return ecs.Length()
        End Function

        Public Sub [set](ByVal i As Int64,
                         ByVal ec As eventComb,
                         ByVal ij As indexedJob,
                         ByVal q As jobqueue,
                         ByVal queue_index As UInt32)
            ecs(i) = ec
            indexedJobs(i) = ij
            jobqueues(i) = q
            queue_indexes(i) = queue_index
        End Sub
    End Class

    Private Sub assert_targetJobs(ByVal jobs As vector(Of pair(Of UInt32, ijob)))
        If isdebugmode() AndAlso jobs.size() > 1 Then
            For i As Int64 = 0 To jobs.size() - 2
                For j As Int64 = i + 1 To jobs.size() - 1
                    assert(object_compare(jobs(i).second, jobs(j).second) = object_compare_undetermined)
                Next
            Next

            For i As Int64 = 0 To jobs.size() - 1
                assert(jobs(i).first >= 0 AndAlso jobs(i).first < targetQueues().size())
            Next
        End If
    End Sub

    Protected NotOverridable Overrides Sub workOn(ByVal job As indexedJob)
        Dim startMs As Int64 = 0
        startMs = Now.milliseconds()
        Dim targets_startMs As Int64 = 0
        Dim targetJobs As vector(Of pair(Of UInt32, ijob)) = Nothing
        Dim waitall As Boolean = False
        targetJobs = New vector(Of pair(Of UInt32, ijob))()
        If distribute(job.job, targetJobs, waitall) Then
            assert_targetJobs(targetJobs)
            Dim targets As targets_t = Nothing
            targets = New targets_t(targetJobs.size())
            For i As Int64 = 0 To targetJobs.size() - 1
                Dim targetIndexedJob As indexedJob = Nothing
                Dim targetQueue As jobqueue = Nothing
                targetQueue = Me.targetQueue(targetJobs(i).first)
                targets.set(i,
                            ijobEventComb(targetJobs(i).second,
                                          targetQueue,
                                          targetIndexedJob),
                            targetIndexedJob,
                            targetQueue,
                            targetJobs(i).first)
            Next

            assert_begin(New eventComb(
                                Function() As Boolean
                                    targets_startMs = Now.milliseconds()
                                    Return (targets.size() = 0 OrElse
                                           ((waitall AndAlso waitfor(targets.ecs)) OrElse
                                            waitone(targets.ecs))) AndAlso
                                           goto_next()
                                End Function,
                                Function() As Boolean
                                    counter.increase(WAIT_TARGET_JOBS_TIME_MS,
                                                     Now.milliseconds() - targets_startMs)
                                    If Not waitall Then
                                        For i As Int64 = 0 To targets.size() - 1
                                            If Not targets.indexedJobs(i).finished() Then
                                                targets.jobqueues(i).update(targets.indexedJobs(i),
                                                                            Sub(ByRef j As indexedJob)
                                                                                j.cancel()
                                                                            End Sub)
                                            End If
                                        Next
                                    End If
                                    Return update(job,
                                                  Sub(ByRef ij As indexedJob)
                                                      If targets.size() > 0 AndAlso
                                                         Not aggregate(targets.indexedJobs,
                                                                       targets.jobqueues,
                                                                       targets.queue_indexes,
                                                                       ij.job()) Then
                                                          raiseError(errorHandle.errorType.warning,
                                                                     "cannot aggregate results")
                                                      End If
                                                      ij.finish()
                                                  End Sub) AndAlso
                                           true_(Sub()
                                                     counter.increase(OVERALL_TIME_MS,
                                                                      Now.milliseconds() - startMs)
                                                 End Sub) AndAlso
                                           goto_end()
                                End Function))
        Else
            assert(update(job))
        End If
    End Sub

    Public Shared Shadows Function initial(ByVal name As String,
                                           ByVal config As configParser,
                                           ByVal distributor As distributor) As Boolean
        assert(Not distributor Is Nothing)
        Dim s As String = Nothing
        If config.getparameter(name, "targetQueues", s) Then
            distributor.initial(s)
            distributor.OVERALL_TIME_MS =
                counter.registerAverageAndLastAverage(strcat(strtoupper(unref(name)), "_OVERALL_TIME_MS"))
            distributor.WAIT_TARGET_JOBS_TIME_MS =
                counter.registerAverageAndLastAverage(strcat(strtoupper(unref(name)), "_WAIT_TARGET_JOBS_TIME_MS"))

            Return consumer.initial(name, config, distributor)
        Else
            raiseError(errorHandle.errorType.user, "no target queues has been set in ", name)
            Return False
        End If
    End Function
End Class
